package xin.marcher.module.inventory.service.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;
import xin.marcher.module.common.constants.inventory.TairInventoryConstant;
import xin.marcher.module.common.core.JsonResult;
import xin.marcher.module.common.enums.inventory.BucketOperateEnum;
import xin.marcher.module.common.enums.inventory.BucketOperateStatusEnum;
import xin.marcher.module.common.enums.inventory.BucketStatusEnum;
import xin.marcher.module.common.enums.inventory.InventoryExceptionCode;
import xin.marcher.module.common.exception.BaseBizException;
import xin.marcher.module.common.exception.CommonErrorCodeEnum;
import xin.marcher.module.common.exception.ProductBizException;
import xin.marcher.module.common.tair.TairCache;
import xin.marcher.module.common.tair.TairLock;
import xin.marcher.module.common.utils.JsonUtil;
import xin.marcher.module.common.utils.SnowflakeIdWorker;
import xin.marcher.module.inventory.cache.InventoryBucketCache;
import xin.marcher.module.inventory.converter.InventoryConverter;
import xin.marcher.module.inventory.domain.bo.*;
import xin.marcher.module.inventory.domain.entity.InventoryBucketConfigDO;
import xin.marcher.module.inventory.domain.request.*;
import xin.marcher.module.inventory.domain.response.BucketCacheDTO;
import xin.marcher.module.inventory.domain.response.ProductInventoryDTO;
import xin.marcher.module.inventory.domain.response.WarningInventoryDTO;
import xin.marcher.module.inventory.generator.core.SegmentNoCache;
import xin.marcher.module.inventory.manage.CalculateThresholdQueue;
import xin.marcher.module.inventory.manage.OperateClearQueue;
import xin.marcher.module.inventory.manage.OperateQueue;
import xin.marcher.module.inventory.mq.producer.BucketClearProducer;
import xin.marcher.module.inventory.mq.producer.BucketOfflineProducer;
import xin.marcher.module.inventory.mq.producer.WarningInventoryProducer;
import xin.marcher.module.inventory.repository.InventoryRepository;
import xin.marcher.module.inventory.repository.LeafAllocNoRepository;
import xin.marcher.module.inventory.service.InventoryBucketService;
import xin.marcher.module.inventory.util.InventorBucketUtil;

import javax.annotation.Resource;
import java.math.BigDecimal;
import java.util.*;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * 库存分桶 业务实现类
 */
@Service
@Slf4j
public class InventoryBucketServiceImpl implements InventoryBucketService {

    @Resource
    private OperateQueue operateQueue;

    @Resource
    private OperateClearQueue operateClearQueue;

    @Resource
    private CalculateThresholdQueue calculateThresholdQueue;

    @Resource
    private InventoryRepository inventoryRepository;

    @Resource
    private InventoryBucketCache inventoryBucketCache;

    @Resource
    private InventoryConverter inventoryConverter;
    @Resource
    private BucketOfflineProducer bucketOfflineProducer;
    @Resource
    private WarningInventoryProducer warningInventoryProducer;
    @Resource
    private BucketClearProducer bucketClearProducer;

    @Resource
    private TairCache tairCache;

    @Resource
    private TairLock tairLock;

    @Resource
    private LeafAllocNoRepository leafAllocNoRepository;

    @Resource
    private SegmentNoCache segmentNoCache;

    /**
     * 库存预警数量
     */
    @Value("${warning.inventory-num}")
    private Integer warningInventoryNum;

    /**
     * 库存预警比例
     */
    @Value("${warning.proportion}")
    private Integer proportion;

    /**
     * 配置的任务内存队列数量
     */
    @Value("${bucket.operate.queue-num:32}")
    private Integer operateQueueNum;

    private ExecutorService executors = Executors.newFixedThreadPool(50);

    /**
     * 库存入桶分配
     *
     * @param request
     */
    @Override
    @Transactional(rollbackFor = Exception.class)
    public void inventorBucket(InventorBucketRequest request) {
        // 1.验证入参必填
        checkInventorParams(request);

        // 锁key
        String key = buildBucketLockKey(request.getSellerId(), request.getSkuId());
        String value = SnowflakeIdWorker.getCode();
        // 注意这里需要锁定中心桶库存
        boolean lock = tairLock.tryLock(key, value);
        // 分配库存的时候，这个卖家的sku是不允许其他相关操作的
        if (lock) {
            try {
                // 2.插入库存入库的记录信息,申请的业务编号有唯一key,避免重复请求
                inventoryRepository.saveInventoryAllotDetail(request);
                // 3.处理数据缓存的操作
                inventoryBucketCache(request);
            } catch (Exception e) {
                log.error("库存初始化失败", e);
            } finally {
                tairLock.unlock(key, value);
            }
        } else {
            throw new BaseBizException("请求繁忙，稍后重试！");
        }
    }


    /**
     * 分桶上线接口
     *
     * @param request
     */
    @Override
    @Transactional(rollbackFor = Exception.class)
    public void bucketOnline(InventorOnlineRequest request) {
        //1.验证入参必填
        checkInventorOnlineParams(request);
        //2. 注意这里需要锁定中心桶库存
        String key = buildBucketLockKey(request.getSellerId(), request.getSkuId());
        String value = SnowflakeIdWorker.getCode();
        boolean lock = tairLock.tryLock(key, value);
        if (lock) {
            try {
                //3. 获取中心桶的库存,并校验是否可上线分桶
                Integer residueNum = checkBucketOnlineNum(key);
                //4.构建新的分桶元数据信息，并写入
                writeBucketCache(request, residueNum);
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                tairLock.unlock(key, value);
            }
        } else {
            throw new BaseBizException("请求繁忙，稍后重试！");
        }
    }

    /**
     * 分桶下线接口
     *
     * @param request
     */
    @Override
    public void bucketOffline(InventorOfflineRequest request) {

        //1.验证入参必填
        checkInventorOfflineParams(request);
        // 过滤只有一个分桶的无效请求
        Boolean isOffline = checkBucketOffline(request);
        if (isOffline) {
//            long start = System.currentTimeMillis();
            //2. 注意这里需要锁定 下线分桶的变更，这个接口默认一次只有一个分桶
            String key = buildBucketOfflineLockKey(request.getSellerId(), request.getSkuId(), request.getBucketNoList().get(0));
            String value = SnowflakeIdWorker.getCode();
            boolean lock = tairLock.tryLock(key, value);
            if (lock) {
                try {
                    //3.先将准备下线的分桶库存从本地和远程列表中移除至不可用列表，避免新的请求进来
                    updateBucketCache(request);
//                    log.error("分桶下线处理时间，下线分桶：{}, 当前时间:{}, 耗时:{}", JSON.toJSONString(request.getBucketNoList()), DateFormatUtil.formatDateTime(new Date()),
//                            System.currentTimeMillis()-start);
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    tairLock.unlock(key, value);
                }
            } else {
                throw new BaseBizException("请求繁忙，稍后重试！");
            }

        }
    }

    /**
     * 判断待下线分桶是否可以下线
     *
     * @param request
     * @return
     */
    private Boolean checkBucketOffline(InventorOfflineRequest request) {
        String key = buildBucketCacheKey(request.getSellerId(), request.getSkuId());
        // 通过远程的缓存判定分桶已被下线，如果回退过程中发生本地缓存更新速度慢影响，则会重试分桶库存下线回源到中心桶上
        String bucketCacheJson = inventoryBucketCache.getBucketCache(key);
        if (!StringUtils.isEmpty(bucketCacheJson)) {
            // 引用缓存组件需要通过通用对象进行对应的缓存获取
            BucketLocalCache bucketLocalCache = JsonUtil.json2Object(bucketCacheJson, BucketLocalCache.class);

            // 可用分桶缓存列表大于1
            if (!Objects.isNull(bucketLocalCache) && bucketLocalCache.getAvailableList().size() > 1) {
                // 并且待下线的分桶在可用分桶列表中
                return bucketLocalCache.getAvailableList().stream()
                        .anyMatch(bucketCache -> request.getBucketNoList().contains(bucketCache.getBucketNo()));
            }
//            log.error("分桶下线失败：{},可用分桶列表：{}", JSON.toJSONString(request.getBucketNoList()), JSON.toJSONString(bucketLocalCache.getAvailableList()));
        }
        return false;
    }


    /**
     * 分桶扩容接口
     *
     * @param bucketCapacity
     */
    @Override
    public void bucketCapacity(BucketCapacity bucketCapacity) {
//        long startTime = System.currentTimeMillis();
        // 获取中心桶库存的库存
        Integer residueNum = getCentreStock(bucketCapacity);
        if (residueNum <= 0) {
            // 中心桶无库存，检查是否触发下线
            checkBucketOffline(bucketCapacity);
            return;
        }

        // 判断本次扩容的分桶，是否有多次扩容失败的情况
        String failNum = tairCache.get(TairInventoryConstant.BUCKET_CAPACITY_FAIL + bucketCapacity.getBucketNo());
        if (StringUtils.isNotBlank(failNum) && Integer.parseInt(failNum) >= 2) {
            // 当前失败次数超过两次了，直接放弃这次扩容,因为失败的太多并且还继续去尝试，会持续的扣减中心桶库存，可能会导致其他可以正常扩容的分桶，没有中心桶库存可以扣减
            return;
        }
        //1. 校验是否已经无需扩容了，如果是则快速结束
        BucketCapacityContext bucketCapacityContext = checkBucketCapacity(bucketCapacity);
        if (!bucketCapacityContext.getIsCapacity()) {
            return;
        }
        // 先锁住中心桶库存，避免此时库存发生变化
        String key = buildBucketLockKey(bucketCapacity.getSellerId(), bucketCapacity.getSkuId());
        String value = SnowflakeIdWorker.getCode();

        boolean lock = tairLock.tryLock(key, value);
        if (lock) {
            try {
                // 再次校验是否需要扩容，此处不允许并发
                bucketCapacityContext = checkBucketCapacity(bucketCapacity);
                if (bucketCapacityContext.getIsCapacity()) {

                    //2.获取中心桶库存的库存
                    residueNum = getCentreStock(bucketCapacity);
                    //3.可以扩容，计算出可回源的库存进行处理
                    if (residueNum > 0) {
                        backSourceInventory(residueNum, bucketCapacityContext);
//                       log.error(bucketCapacity.getBucketNo()+"处理扩容消耗时间{}",System.currentTimeMillis() - startTime );
                    } else {
                        //4. 中心桶无库存，检查是否触发下线
                        checkBucketOffline(bucketCapacity);
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                tairLock.unlock(key, value);
            }
        } else {
            throw new BaseBizException("请求繁忙，稍后重试！");
        }
    }

    /**
     * 查询某个卖家的商品相关库存信息
     *
     * @param request
     * @return
     */
    @Override
    public ProductInventoryDTO queryProductInventoryList(InventoryRequest request) {
        return querySellerProductBucketList(request.getSellerId(), request.getSkuId());
    }

    /**
     * 清空分桶库存，分桶库存放回中央库存
     *
     * @param request
     */
    @Override
    public void bucketClear(BucketClearRequest request) {
        long start = System.currentTimeMillis();
        String key = buildBucketCacheKey(request.getSellerId(), request.getSkuId());
        String bucketCache = inventoryBucketCache.getBucketCache(key);
        if (!StringUtils.isEmpty(bucketCache)) {
            // 引用缓存组件需要通过通用对象进行对应的缓存获取
            BucketLocalCache bucketLocalCache = JsonUtil.json2Object(bucketCache, BucketLocalCache.class);

            updateBucketInventory(request, bucketLocalCache);
        }
//        log.error("清空下线分桶库存:{}，时间：{}", JSON.toJSONString(request.getBucketNoList()), System.currentTimeMillis()-start);

        //商品库存值预警
        warningProductInventory(bucketCache);
    }

    /**
     * 执行分桶操作任务
     *
     * @return
     */
    @Override
    public JsonResult processBucketOperate() {
        String code = SnowflakeIdWorker.getCode();
        boolean lock = tairLock.lock(TairInventoryConstant.BUCKET_OPERATE_PROCESS, code);
        if (lock) {
            try {
                int page = 1;
                // 分页获取要执行的操作列表
                List<InventoryBucketOperateBO> inventoryBucketOperateBOS =
                        inventoryRepository.queryBucketOperateList(BucketOperateStatusEnum.UN_PROCESS.getCode(), page, operateQueueNum);
                while (!CollectionUtils.isEmpty(inventoryBucketOperateBOS)) {
                    // 当前有多少个任务，添加至缓存中,每执行完一个任务后，会
                    inventoryBucketCache.incrOperateCount(TairInventoryConstant.BUCKET_OPERATE_PROCESS_COUNT,
                            inventoryBucketOperateBOS.size());
                    // 将任务轮询添加到任务执行队列中
                    for (InventoryBucketOperateBO inventoryBucketOperateBO : inventoryBucketOperateBOS) {
                        operateQueue.offerByRoundRobin(inventoryBucketOperateBO);
                    }

                    // 分页获取下一页要执行的任务列表
                    inventoryBucketOperateBOS = inventoryRepository.queryBucketOperateList(BucketOperateStatusEnum.UN_PROCESS.getCode(), ++page, operateQueueNum);
                }
                // 等待本次所有任务执行完成，这里不能按照队列的大小来判断，因为可能队列中有最后一个任务，刚好取出来这个任务，但是没执行，这时队列大小也为0了
                while (inventoryBucketCache.getOperateCount(TairInventoryConstant.BUCKET_OPERATE_PROCESS_COUNT) != 0) {
                    try {
                        Thread.sleep(200);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }

                // 所有任务都执行过了之后，获取id
                List<Long> ids = inventoryBucketCache.getOperateId(TairInventoryConstant.BUCKET_OPERATE_PROCESS);
                if (!CollectionUtils.isEmpty(ids)) {
                    log.info("修改任务状态为已完成：{}", JSON.toJSONString(ids));
                    inventoryRepository.bucketOperateSuccess(ids, BucketOperateStatusEnum.FINISH.getCode());
                }
            } finally {
                inventoryBucketCache.removeOperateId(TairInventoryConstant.BUCKET_OPERATE_PROCESS);
                tairLock.unlock(TairInventoryConstant.BUCKET_OPERATE_PROCESS, code);
            }
        }
        return JsonResult.buildSuccess();
    }

    /**
     * 清除执行完的分桶操作任务
     *
     * @return
     */
    @Override
    public JsonResult bucketOperateFinishedClear() {
        String code = SnowflakeIdWorker.getCode();
        boolean lock = tairLock.lock(TairInventoryConstant.BUCKET_OPERATE_CLEAR_PROCESS, code);
        if (lock) {
            try {
                int page = 1;
                List<InventoryBucketOperateBO> inventoryBucketOperateBOS =
                        inventoryRepository.queryBucketOperateList(BucketOperateStatusEnum.FINISH.getCode(), page, operateQueueNum);
                while (!CollectionUtils.isEmpty(inventoryBucketOperateBOS)) {
                    // 当前有多少个任务，添加至缓存中,每执行完一个任务后，会
                    inventoryBucketCache.incrOperateCount(TairInventoryConstant.BUCKET_OPERATE_CLEAR_PROCESS_COUNT,
                            inventoryBucketOperateBOS.size());
                    // 将任务轮询添加到任务执行队列中
                    for (InventoryBucketOperateBO inventoryBucketOperateBO : inventoryBucketOperateBOS) {
                        operateClearQueue.offerByRoundRobin(inventoryBucketOperateBO);
                    }

                    // 分页获取下一页要执行的任务列表
                    inventoryBucketOperateBOS = inventoryRepository.queryBucketOperateList(BucketOperateStatusEnum.FINISH.getCode(), ++page, operateQueueNum);
                }

                // 等待本次所有任务执行完成
                while (inventoryBucketCache.getOperateCount(TairInventoryConstant.BUCKET_OPERATE_CLEAR_PROCESS_COUNT) != 0) {
                    try {
                        Thread.sleep(200);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }

                // 删除已经执行完成的任务
                List<Long> ids = inventoryBucketCache.getOperateId(TairInventoryConstant.BUCKET_OPERATE_CLEAR_PROCESS);
                if (!CollectionUtils.isEmpty(ids)) {
                    log.info("清除已完成的任务：{}", JSON.toJSONString(ids));
                    inventoryRepository.deleteBatchBucketOperate(ids);
                }
            } finally {
                inventoryBucketCache.removeOperateId(TairInventoryConstant.BUCKET_OPERATE_CLEAR_PROCESS);
                tairLock.unlock(TairInventoryConstant.BUCKET_OPERATE_CLEAR_PROCESS, code);
            }
        }
        return JsonResult.buildSuccess();
    }

    /**
     * 清除执行完的分桶操作任务
     *
     * @param operate
     */
    @Override
    public void clearBucketOperate(InventoryBucketOperateBO operate) {
//        Long startTime = System.currentTimeMillis();
        List<InventoryOperateFailBO> inventoryOperateFailBOS = inventoryRepository.queryFailOperateList(operate.getOperateId());

        Map<String, InventoryOperateFailBO> operateFailBOMap = inventoryOperateFailBOS.stream()
                .collect(Collectors.toMap(InventoryOperateFailBO::getBucketNo, Function.identity()));

        // 先处理缓存中存储的失败记录
        processCacheRollback(operate, operateFailBOMap);

        // 再处理缓存中没有存储的但是DB中存储的失败记录
        processDbRollback(operate, operateFailBOMap);

        if (!CollectionUtils.isEmpty(inventoryOperateFailBOS)) {
            // 所有的都执行完成了，删除处理完成的数据库记录，缓存记录在处理过程中已经执行了
            inventoryRepository.deleteBatchOperateFail(inventoryOperateFailBOS);
        }

        // 执行成功的，删除记录
        tairCache.delete(TairInventoryConstant.BUCKET_OPERATE_PREFIX + operate.getId());
        inventoryBucketCache.addOperateId(TairInventoryConstant.BUCKET_OPERATE_CLEAR_PROCESS, operate.getId());
        // 任务执行后，减少任务数量
        inventoryBucketCache.decrOperateCount(TairInventoryConstant.BUCKET_OPERATE_CLEAR_PROCESS_COUNT);
//        log.error("本次分桶清除任务耗时{},任务对象{}", System.currentTimeMillis() - startTime, JSONObject.toJSONString(operate));
    }

    /**
     * 计算分桶某个时间区间的库存具体消费速度，生成预估的下线阈值
     */
    @Override
    public void calculateOfflineThreshold() {
        //1. 先获取到目前售卖中的缓存商品key
        List<BucketLocalCache> bucketLocalCacheList = inventoryBucketCache.getBucketLocalCacheList();
        if (!CollectionUtils.isEmpty(bucketLocalCacheList)) {
            // 将商品缓存key加入到队列中，计算出每个商品的预估下线阈值
            for (BucketLocalCache bucketLocalCache : bucketLocalCacheList) {
                calculateThresholdQueue.offerByRoundRobin(bucketLocalCache);
            }
        }
    }

    /**
     * 处理数据库中的失败回退
     *
     * @param operate
     * @param operateFailBOMap
     */
    private void processDbRollback(InventoryBucketOperateBO operate, Map<String, InventoryOperateFailBO> operateFailBOMap) {
        // 获取中心桶库存的key
        String key = buildSellerInventoryKey(operate.getSellerId(), operate.getSkuId());
        // 如果缓存中数据操作完，数据库中还有未处理的，需要处理数据库中的
        if (!CollectionUtils.isEmpty(operateFailBOMap)) {
            for (Map.Entry<String, InventoryOperateFailBO> entry : operateFailBOMap.entrySet()) {
                InventoryOperateFailBO operateFailBO = entry.getValue();
                if (TairInventoryConstant.OPERATE_SET_FAIL.equals(operateFailBO.getFailType())) {
                    // 这里是分桶库存初始化失败了，需要将数据放回中心桶
                    tairCache.incr(key,
                            operateFailBO.getInventoryNum());
                    // 处理成功，将失败列表中的数据删除
                    tairCache.removeJsonExhset(TairInventoryConstant.OPERATE_SET_FAIL, operateFailBO.getBucketNo(),
                            operate.getOperateId());
                } else if (TairInventoryConstant.OPERATE_INCR_FAIL.equals(operateFailBO.getFailType())) {
                    // 这里是分桶库存增加失败了，需要将数据放回中心桶
                    tairCache.incr(key,
                            operateFailBO.getInventoryNum());
                    // 处理成功，将失败列表中的数据删除
                    tairCache.removeJsonExhset(TairInventoryConstant.OPERATE_INCR_FAIL, operateFailBO.getBucketNo(),
                            operate.getOperateId());

                    // 处理完回退之后，如果是扩容的，需要将扩容失败次数减一
                    if (BucketOperateEnum.CAPACITY.getCode().equals(operate.getOperateType())) {
                        // 扩容的时候，一次操作只有一个分桶数据变更，所以这里跟缓存中获取的失败记录，最多执行一次，不会导致这次失败的任务，将其他任务的次数减去了
                        tairCache.decr(TairInventoryConstant.BUCKET_CAPACITY_FAIL + operateFailBO.getBucketNo(), 1);
                    }
                }
            }
        }
    }

    /**
     * 处理缓存中的失败回退
     *
     * @param operate
     * @param operateFailBOMap
     */
    private void processCacheRollback(InventoryBucketOperateBO operate, Map<String, InventoryOperateFailBO> operateFailBOMap) {
        List<BucketCacheBO> bucketCacheBOS = JSON.parseArray(operate.getBucket(), BucketCacheBO.class);
        // 获取中心桶库存的key
        String key = buildSellerInventoryKey(operate.getSellerId(), operate.getSkuId());

        for (BucketCacheBO bucketCacheBO : bucketCacheBOS) {
            // 分桶的缓存操作有两个，一个是初始化设置库存，一个是回源增加库存

            // 获取初始化库存失败的操作
            Object bucketNum = tairCache.getJsonExhset(TairInventoryConstant.OPERATE_SET_FAIL,
                    bucketCacheBO.getBucketNo(), operate.getOperateId());
            if (!Objects.isNull(bucketNum)) {
                // 这里是分桶库存初始化失败了，需要将数据放回中心桶
                tairCache.incr(key, (Integer) bucketNum);
                // 处理成功，将失败列表中的数据删除
                tairCache.removeJsonExhset(TairInventoryConstant.OPERATE_SET_FAIL, bucketCacheBO.getBucketNo(),
                        operate.getOperateId());
                operateFailBOMap.remove(bucketCacheBO.getBucketNo());
            }

            // 获取增加库存失败的操作
            bucketNum = tairCache.getJsonExhset(TairInventoryConstant.OPERATE_INCR_FAIL, bucketCacheBO.getBucketNo(), operate.getOperateId());
            if (!Objects.isNull(bucketNum)) {
                // 这里是分桶库存增加失败了，需要将数据放回中心桶
                tairCache.incr(key, (Integer) bucketNum);
                // 处理成功，将失败列表中的数据删除
                tairCache.removeJsonExhset(TairInventoryConstant.OPERATE_INCR_FAIL, bucketCacheBO.getBucketNo(),
                        operate.getOperateId());
                operateFailBOMap.remove(bucketCacheBO.getBucketNo());

                // 处理完回退之后，如果是扩容的，需要将扩容失败次数减一
                if (BucketOperateEnum.CAPACITY.getCode().equals(operate.getOperateType())) {
                    tairCache.decr(TairInventoryConstant.BUCKET_CAPACITY_FAIL + bucketCacheBO.getBucketNo(), 1);
                }
            }
        }

    }

    /**
     * 处理下线任务
     *
     * @param inventoryBucketOperateBO
     */
    @Override
    public void processOfflineOperate(InventoryBucketOperateBO inventoryBucketOperateBO) {
//        log.error("执行分桶下线{}",JSONObject.toJSONString(inventoryBucketOperateBO) );
        log.info("发送分桶清空的消息：{}", inventoryBucketOperateBO.getBucket());
        BucketLocalCache bucketLocalCache = JSON.parseObject(inventoryBucketOperateBO.getFeature(), BucketLocalCache.class);
        List<BucketCacheBO> bucketCacheBOList = JSON.parseArray(inventoryBucketOperateBO.getBucket(), BucketCacheBO.class);
        List<String> bucketCacheList = bucketCacheBOList.stream().map(BucketCacheBO::getBucketNo).collect(Collectors.toList());
        BucketClearRequest bucketClearRequest = new BucketClearRequest(bucketLocalCache.getSkuId(),
                bucketLocalCache.getSellerId(), bucketCacheList, 0);
        for (int i = 0; i < 3; i++) {
            try {
                // 发送清空下线分桶库存的消息，默认这里不存在需要处理的中心桶库存
                bucketClearProducer.sendBucketClear(bucketClearRequest);
                // 发送成功，直接返回，发送失败，继续处理
                return;
            } catch (Exception e) {
                // 这里消息发送失败了,再次尝试发送，失败三次的话，就抛出异常
            }
        }

        // 到这里就是消息多次发送失败，需要做其他的处理
        throw new ProductBizException(CommonErrorCodeEnum.SEND_MQ_FAILED);

    }

    /**
     * 处理扩容任务
     *
     * @param operate
     */
    @Override
    public void processCapacityOperate(InventoryBucketOperateBO operate) {
//       Long startTime = System.currentTimeMillis();
        List<BucketCacheBO> bucketCacheBOList = JSON.parseArray(operate.getBucket(), BucketCacheBO.class);
        // 回源分桶的库存
        String bucketNo = bucketCacheBOList.get(0).getBucketNo();
        Integer incr = tairCache.retryIncr(bucketNo, operate.getInventoryNum(), 3);

        if (incr >= 0) {
            // 执行成功，从失败列表中移除
            // 移除数据库
            inventoryRepository.deleteBucketOperate(operate.getId());
            // 移除缓存
            tairCache.removeJsonExhset(TairInventoryConstant.OPERATE_INCR_FAIL, bucketNo, operate.getOperateId());
            // 回源成功后将失败次数减一
            tairCache.decr(TairInventoryConstant.BUCKET_CAPACITY_FAIL + bucketCacheBOList.get(0).getBucketNo(), 1);
        }
//        log.error("执行扩容操作{},本次耗时{},执行结果{}",JSONObject.toJSONString(operate),System.currentTimeMillis() - startTime, incr >=0 ? true : false);
    }


    /**
     * 校验本次请求是否还需要执行扩容处理
     *
     * @param bucketCapacity
     * @return
     */
    private BucketCapacityContext checkBucketCapacity(BucketCapacity bucketCapacity) {
        String key = bucketCapacity.getSellerId() + bucketCapacity.getSkuId();
        //1.获取远程的分桶缓存信息
        Integer residueNum = getBucketInventoryNum(bucketCapacity.getBucketNo());
        //2.获取 缓存元数据信息
        BucketLocalCache bucketLocalCache = inventoryBucketCache.getBucketLocalCache(key);
        //3.校验是否还需要执行扩容
        List<BucketCacheBO> availableList = bucketLocalCache.getAvailableList();
        InventoryBucketConfigDO inventoryBucketConfig = bucketLocalCache.getInventoryBucketConfig();
        if (!CollectionUtils.isEmpty(availableList)) {
            for (BucketCacheBO bucketCacheBO : availableList) {
                // 具体使用的哪个分桶扣减库存
                if (bucketCacheBO.getBucketNo().equals(bucketCapacity.getBucketNo())) {
                    // 当前分桶的分配总库存
                    Integer bucketNum = bucketCacheBO.getBucketNum();
                    // 触发回源比例的百分比
                    Integer backSourceProportion = inventoryBucketConfig.getBackSourceProportion();
                    int backSourceNum = bucketNum * backSourceProportion / 100;
                    // 回源比例的库存 大于剩余的库存，触发异步扩容
                    return new BucketCapacityContext(residueNum, backSourceNum > residueNum, bucketCapacity);
                }
            }
        }
        inventoryBucketCache.threadLocalRemove();
        // 如果不在可用列表里面，则意味已下线，快速结束掉
        return new BucketCapacityContext(residueNum, false, bucketCapacity);
    }

    /**
     * 回源库存到分桶上
     *
     * @param residueNum            中心桶库存
     * @param bucketCapacityContext 扩容上下文对象
     */
    private void backSourceInventory(Integer residueNum, BucketCapacityContext bucketCapacityContext) {
        // 首先需要当前分桶的库存，其次还需要获取目前分桶的可发库存深度(第一次初始化的时候分配的库存)
        // 根据当初分配的库存深度以及最大库存深度以及中心桶库存，得出均匀到目前支持可用的分桶均匀分配库存大概数量，
        // 同时根据本次同步的库存数量刷新分桶的实际库存深度
        BucketCapacity bucketCapacity = bucketCapacityContext.getBucketCapacity();
        //1. 先获取本地的分桶元数据信息，获取当前分桶的总 发放上限
        String key = bucketCapacity.getSellerId() + bucketCapacity.getSkuId();
        //中心桶库存key
        String sellerInventoryKey = buildSellerInventoryKey(bucketCapacity.getSellerId(), bucketCapacity.getSkuId());
        BucketLocalCache bucketLocalCache = inventoryBucketCache.getBucketLocalCache(key);
        String bucketNo = "";
        try {
            InventoryBucketConfigDO inventoryBucketConfig = bucketLocalCache.getInventoryBucketConfig();
            List<BucketCacheBO> availableList = bucketLocalCache.getAvailableList();
            Integer inventoryNum = 0;
            // 获取实际配置的最大可用库存深度
            Integer maxBucketNum = availableList.stream().mapToInt(BucketCacheBO::getBucketNum).sum();
            BucketCacheBO bucketCache = null;
            for (BucketCacheBO bucketCacheBO : availableList) {
                if (bucketCacheBO.getBucketNo().equals(bucketCapacity.getBucketNo())) {
                    bucketCache = bucketCacheBO;
                    bucketNo = bucketCache.getBucketNo();
                    break;
                }
            }
            // 这里没有匹配到分桶，则该分桶已被下线，不处理后续流程
            if (Objects.isNull(bucketCache)) {
                return;
            }
            // 中心桶库存超过最大深度库存(全部分桶总计)，直接以配置的回源步长增长库存
            if (residueNum > maxBucketNum) {
                inventoryNum = inventoryBucketConfig.getBackSourceStep();
            } else {
                inventoryNum = calcEvenInventoryNum(maxBucketNum, inventoryBucketConfig, residueNum, bucketCache);
            }
            // 填充变更元数据关于库存的深度数据，
            Integer maxDepthNum = getMaxDepthNum(inventoryNum, inventoryBucketConfig, bucketCache, bucketCapacityContext);

            // 更新分桶元数据相关信息
            refreshBucketCache(maxDepthNum, bucketLocalCache, bucketCapacity.getBucketNo(), inventoryNum);

            // 扣减中心桶库存
            Integer decr = tairCache.decr(sellerInventoryKey, inventoryNum);
            if (decr < 0) {
                // 中心桶扣减失败，直接返回
                throw new BaseBizException(InventoryExceptionCode.INVENTORY_CACHE);
            }
            // 回源分桶的库存
            Integer incr = tairCache.retryIncr(bucketCache.getBucketNo(), inventoryNum, 3);
            if (incr < 0) {
                // 这里扣减中心桶成功了，但是回源分桶库存失败了，记录失败的操作，重试，如果重试失败了，则库存返回中心桶

                // 本次操作的唯一id
                String operateId = SnowflakeIdWorker.getCode();
                // 分桶信息入库
                inventoryRepository.saveBucketDetail(operateId, bucketLocalCache, BucketOperateEnum.CAPACITY.getCode(),
                        Lists.newArrayList(bucketCache), inventoryNum);

                // 考虑到这里同一个分桶，可能会有多个任务没处理完，value值需要用对象来存储
                inventoryRepository.saveFailOperate(operateId, TairInventoryConstant.OPERATE_INCR_FAIL,
                        bucketCache.getBucketNo(), inventoryNum);

                tairCache.appendJsonExhset(TairInventoryConstant.OPERATE_INCR_FAIL, bucketCache.getBucketNo(),
                        operateId, inventoryNum);

                throw new BaseBizException(InventoryExceptionCode.INVENTORY_CACHE);
            }
//            log.error("本次分桶:{}，回源库存：{}, 回源后分桶库存：{}， 中心桶剩余库存：{}", bucketCache.getBucketNo(),inventoryNum, incr, decr);
        } catch (Exception e) {
            // 增加分桶扩容失败次数，如果次数超过两次了，则不会尝试更新缓存，直接写DB，让DB去操作
            // DB操作成功，会减少失败次数
            tairCache.incr(TairInventoryConstant.BUCKET_CAPACITY_FAIL + bucketNo);
            throw new BaseBizException(InventoryExceptionCode.INVENTORY_CACHE);
        } finally {
            inventoryBucketCache.threadLocalRemove();
        }
    }

    /**
     * 计算出均匀后的每个分库实际 分配的库存值
     *
     * @param maxBucketNum          最大的库存深度
     * @param inventoryBucketConfig 分桶配置
     * @param residueNum            中心桶剩余库存
     * @param bucketCache           扩容分桶
     * @return
     */
    private Integer calcEvenInventoryNum(Integer maxBucketNum,
                                         InventoryBucketConfigDO inventoryBucketConfig,
                                         Integer residueNum,
                                         BucketCacheBO bucketCache) {
        // 获取当前扩容的分桶深度
        Integer bucketDepthNum = bucketCache.getBucketNum();
        // 得到扩容的分桶深度 和当前全部可用分桶的库存深度，计算占比为，根据占比计算出 回源的步长，注意最小深度，如果计算后的步长小于最小库存深度，则默认取最小库存深度
        BigDecimal proportion = new BigDecimal(bucketDepthNum).divide(new BigDecimal(maxBucketNum), 6, BigDecimal.ROUND_DOWN);
        // 根据比例计算出可分配的库存
        BigDecimal allotNum = new BigDecimal(residueNum).multiply(proportion).setScale(0, BigDecimal.ROUND_DOWN);

        if (allotNum.compareTo(new BigDecimal(inventoryBucketConfig.getMinDepthNum())) < 0) {
            allotNum = new BigDecimal(inventoryBucketConfig.getMinDepthNum());
        }
        // 当最小深度都已无法满足剩余库存，则以实际剩余库存扩容
        if (new BigDecimal(residueNum).compareTo(allotNum) < 0) {
            return residueNum;
        }
        // 得到扩容的库存信
        return allotNum.intValue();
    }

    /**
     * 刷新分桶元数据缓存
     *
     * @param maxDepthNum      分桶最大库存深度
     * @param bucketLocalCache 分桶元数据信息
     * @param bucketNo         分桶编号
     */
    private void refreshBucketCache(Integer maxDepthNum, BucketLocalCache bucketLocalCache, String bucketNo, Integer inventoryNum) {
        List<BucketCacheBO> availableList = bucketLocalCache.getAvailableList();
        for (BucketCacheBO bucketCacheBO : availableList) {
            if (bucketCacheBO.getBucketNo().equals(bucketNo)) {
                // 每次库存具体深度变化都要更细，否则很容易触发 回源的比例
                bucketCacheBO.setBucketNum(maxDepthNum);
                bucketCacheBO.setAllotNum(inventoryNum + (Objects.isNull(bucketCacheBO.getAllotNum()) ? 0 :
                        bucketCacheBO.getAllotNum()));
                break;
            }
        }
        String key = buildBucketCacheKey(bucketLocalCache.getSellerId(), bucketLocalCache.getSkuId());
        //1. 刷新本地缓存
        inventoryBucketCache.setBucketLocalCache(key, bucketLocalCache);
    }

    /**
     * 返回目前扩容后的库存深度，库存深度只允许增长不允许减少
     *
     * @param inventoryNum          步长扩容库存
     * @param inventoryBucketConfig 分桶配置信息
     * @param bucketCache           分桶信息
     * @return
     */
    private Integer getMaxDepthNum(Integer inventoryNum, InventoryBucketConfigDO inventoryBucketConfig, BucketCacheBO bucketCache, BucketCapacityContext bucketCapacityContext) {
        //1. 获取当前分桶的实际库存,实际库存和真实库存会有差异，但是这里只是计算一个大概库存深度，无需精确
        Integer residueNum = bucketCapacityContext.getResidueNum();
        // 预估出实际库存深度， 当前（分桶库存+步长增长库存）
        Integer maxBucketNum = residueNum + inventoryNum;
        if (bucketCache.getBucketNum() > maxBucketNum) {
            return bucketCache.getBucketNum();
        }
        // 实际库存深度，不能超过配置的最大库存深度，同理，最小深度也不能小于最小的库存深度
        if (inventoryBucketConfig.getMaxDepthNum() < maxBucketNum) {
            return inventoryBucketConfig.getMaxDepthNum();
        }
        return maxBucketNum;
    }

    /**
     * 校验当前分桶是否触发下线的阈值
     *
     * @param bucketCapacity
     */
    private void checkBucketOffline(BucketCapacity bucketCapacity) {
//        long startTime = System.currentTimeMillis();
        //1. 获取当前分桶的配置信息
        String key = bucketCapacity.getSellerId() + bucketCapacity.getSkuId();
        BucketLocalCache bucketLocalCache = inventoryBucketCache.getBucketLocalCache(key);
        //2.检测分桶的库存是否触发下线阈值，先获取当前分桶的具体库存以及下线配置阈值
        Integer thresholdValue = getThresholdValue(bucketLocalCache);
        Integer inventoryNum = getBucketInventoryNum(bucketCapacity.getBucketNo());
        //3.如触发下线，发生消息调用分桶下线
        if (thresholdValue > inventoryNum) {
            if (bucketLocalCache.getAvailableList().stream().anyMatch(bucketCacheBO -> bucketCapacity.getBucketNo().equals(bucketCacheBO.getBucketNo()))) {
//                log.error("触发下线{}，阈值{},触发分桶{},当前库存值{},耗时{}", thresholdValue > inventoryNum,thresholdValue,bucketCapacity.getBucketNo(),inventoryNum,System.currentTimeMillis() - startTime);
                sendAsynchronous(bucketCapacity);
            }
        }
        inventoryBucketCache.threadLocalRemove();
    }

    /**
     * 获取到商品的预估下线阈值
     *
     * @param bucketLocalCache
     * @return
     */
    private Integer getThresholdValue(BucketLocalCache bucketLocalCache) {
        String thresholdKey = buildSellerInventoryThresholdKey(bucketLocalCache.getSellerId(), bucketLocalCache.getSkuId());
        String thresholdValue = tairCache.get(thresholdKey);
        if (StringUtils.isNotEmpty(thresholdValue)) {
            return Integer.valueOf(thresholdValue);
        }
        return bucketLocalCache.getInventoryBucketConfig().getThresholdValue();
    }

    /**
     * 获取得到 当前分桶对应的实际剩余库存
     *
     * @param bucketNo
     * @return
     */
    private Integer getBucketInventoryNum(String bucketNo) {

        String bucketNum = tairCache.get(bucketNo);
        if (StringUtils.isEmpty(bucketNum)) {
            return 0;
        }
        return Integer.valueOf(bucketNum);
    }

    /**
     * 对分桶进行异步下线
     *
     * @param bucketCapacity
     */
    private void sendAsynchronous(BucketCapacity bucketCapacity) {
        //1. 构建分桶下线接口模型
        InventorOfflineRequest offlineRequest = buildOfflineBucketInfo(bucketCapacity);
        //2.发送消息，通知处理分桶下线
        bucketOfflineProducer.sendBucketOffline(offlineRequest);
    }

    /**
     * 构建分桶下线的消息发生模型
     *
     * @param bucketCapacity 分桶下线的对象
     */
    private InventorOfflineRequest buildOfflineBucketInfo(BucketCapacity bucketCapacity) {
        InventorOfflineRequest offlineRequest = inventoryConverter.converter(bucketCapacity);
        List<String> bucketNoList = new ArrayList<>();
        bucketNoList.add(bucketCapacity.getBucketNo());

        offlineRequest.setBucketNoList(bucketNoList);
        return offlineRequest;
    }

    /**
     * 返回中心桶库存，如返回中心库存大于0则允许处理
     *
     * @param bucketCapacity
     * @return
     */
    private Integer getCentreStock(BucketCapacity bucketCapacity) {
        // 获取中心桶库存的key
        String key = buildSellerInventoryKey(bucketCapacity.getSellerId(), bucketCapacity.getSkuId());

        String centreCacheNum = tairCache.get(key);
        if (!StringUtils.isEmpty(centreCacheNum)) {
            return Integer.valueOf(centreCacheNum);
        }
        return 0;
    }

    /**
     * 对商品的库存发生变化进行预警处理
     *
     * @param bucketCache   分桶缓存信息
     */
    private void warningProductInventory(String bucketCache) {
        //1. 批量获取一下可用的缓存分桶列表编号
        BucketLocalCache bucketLocalCache = JsonUtil.json2Object(bucketCache, BucketLocalCache.class);
        if (Objects.isNull(bucketLocalCache)) {
            return;
        }
        List<BucketCacheBO> availableList = bucketLocalCache.getAvailableList();
        //2.批量获取汇总商品剩余库存(分桶下线代表中心桶库存已经没有了，不校验中心桶库存)
        List<String> cacheKeyList = availableList.stream().map(BucketCacheBO::getBucketNo).collect(Collectors.toList());
        List<String> productInventoryList = tairCache.mget(cacheKeyList);
        //3.检测卖家单个商品的总库存 是否触发最小值预警或者百分比预警，是则异步消息通知供需服务
        Integer sumInventoryNum = 0;
        for (int i = 0; i < productInventoryList.size(); i++) {
            String inventoryNum = productInventoryList.get(i);
            if (StringUtils.isNotEmpty(inventoryNum)) {
                sumInventoryNum = sumInventoryNum + Integer.valueOf(inventoryNum);
            }
        }
        Boolean isWarning = false;
        // 如果实际库存值，小于预警值，或者总库存触发比例阈值，异步消息通知
        if (sumInventoryNum < warningInventoryNum) {
            isWarning = true;
        }
        // 未触发最小库存预警，检测是否触发最小比例预警
        if (!isWarning) {
            // 总的库存深度，不仅仅要看可用分桶的库存深度，还要看下线的库存深度，从而计算出一个当时实际分配的库存深度，计算出一个预警值
            int sumBucketNum = availableList.stream().mapToInt(BucketCacheBO::getBucketNum).sum();
            List<BucketCacheBO> undistributedList = bucketLocalCache.getUndistributedList();
            if (!CollectionUtils.isEmpty(undistributedList)) {
                sumBucketNum = sumBucketNum + undistributedList.stream().mapToInt(cacheBO -> Objects.isNull(cacheBO.getBucketNum()) ? 0 : cacheBO.getBucketNum()).sum();
            }
            // 预警比例
            BigDecimal warningProportion = new BigDecimal(proportion).divide(new BigDecimal(100), 3, BigDecimal.ROUND_DOWN);
            // 库存占比
            BigDecimal inventoryProportion = new BigDecimal(sumInventoryNum).divide(new BigDecimal(sumBucketNum), 6, BigDecimal.ROUND_HALF_UP);
            // 配置的预警比例，大于分配的实际库存深度和已剩的库存占比
            if (warningProportion.compareTo(inventoryProportion) > 0) {
                isWarning = true;
            }
        }
        // 异步消息通知预警
        if (isWarning) {
            WarningInventoryDTO warningInventoryDTO = inventoryConverter.converterDTO(bucketLocalCache);
            warningInventoryProducer.sendWarningInventory(warningInventoryDTO);
        }
    }

    /**
     * 将分桶的缓存库存返回给中心桶库存上
     *
     * @param request
     * @param bucketLocalCache
     */
    private void updateBucketInventory(BucketClearRequest request, BucketLocalCache bucketLocalCache) {
        // 中心桶的库存key
        String key = buildSellerInventoryKey(bucketLocalCache.getSellerId(), bucketLocalCache.getSkuId());
        // 中心桶需要回源的库存，默认为0
        Integer inventoryNum = request.getInventoryNum();
        // 准备操作下线的分桶
        List<String> bucketCacheList = request.getBucketNoList();
        // 下线的分桶列表
        List<String> undistributedList = bucketLocalCache.getUndistributedList().stream().map(BucketCacheBO::getBucketNo).collect(Collectors.toList());

        // 只处理已经下线的分桶
        bucketCacheList = bucketCacheList.stream().filter(undistributedList::contains).collect(Collectors.toList());
        // 当分桶状态不是已下线的状态，验证这个分桶是否需要处理为下线状态
        if (CollectionUtils.isEmpty(bucketCacheList)) {
            BucketCapacity bucketCapacity = inventoryConverter.converter(request);
            for (String bucketNo : request.getBucketNoList()) {
                bucketCapacity.setBucketNo(bucketNo);
                // 再次校验分桶是否触发下线
                checkBucketOffline(bucketCapacity);
            }
            // 不处理分桶库存回退，但是如果有中心桶库存需要重试，这里进行重试
            bucketCacheList = new ArrayList<>();
        }
        // 标记处理过程中失败的数据，如果是缓存没有库存这种是不会加入
        List<String> failureBucketCacheList = new ArrayList<>();

        for (String bucketNo : bucketCacheList) {
            // 先获取下线的分桶实际剩余库存
            String bucketNum = tairCache.get(bucketNo);
            // 当分桶的库存大于0的时候才处理
            if (!StringUtils.isEmpty(bucketNum) && Integer.valueOf(bucketNum) > 0) {
                // 清理下线的分桶库存,设置为0
                Integer result = tairCache.decr(bucketNo, Integer.parseInt(bucketNum));
                if (result >= 0) {
//                    log.error("下线分桶，bucketNo：{}，desc:{}", bucketNo, bucketNum);
                    inventoryNum = inventoryNum + Integer.parseInt(bucketNum);
                } else {
//                    log.error("分桶已下线，bucketNo：{}", bucketNo);
                    failureBucketCacheList.add(bucketNo);
                }
            }
        }
        if (inventoryNum > 0) {
            // 将下线的剩余库存加至 中心桶库存上
            Integer incr = tairCache.retryIncr(key, inventoryNum, 3);
            // 当返回的值大于0，则意味本次操作回源中心桶库存成功，标记为0，当存在失败的分桶下线不会累计添加上次任务处理的中心桶库存
//            log.error("回源中心桶，inventoryNum：{}, after value :{}", inventoryNum, incr);
            if (incr >= 0) {
                inventoryNum = 0;
            }
        }
        // 对本次库存操作失败的分桶信息，重新写入MQ进行重试，这里只有回退库存失败的或者中心桶库存没有回源成功的才会再次发送，如果是已经被扣减掉库存的查询的时候会过滤掉
        if (!CollectionUtils.isEmpty(failureBucketCacheList) || inventoryNum > 0) {
            // 发送清空下线分桶库存的消息
            bucketClearProducer.sendBucketClear(new BucketClearRequest(bucketLocalCache.getSkuId(), bucketLocalCache.getSellerId(), failureBucketCacheList, inventoryNum));
        }
    }

    /**
     * 移除本地分桶的对应分桶列表以及远程的分桶列表
     *
     * @param request 下线的请求参数列表
     */
    private void updateBucketCache(InventorOfflineRequest request) {
        //下线的分桶列表
        List<String> bucketCacheList = request.getBucketNoList();

        String key = buildBucketCacheKey(request.getSellerId(), request.getSkuId());
        //1. 获取到本地的缓存列表
        BucketLocalCache bucketLocalCache = inventoryBucketCache.getBucketLocalCache(request.getSellerId() + request.getSkuId());
        try {
            //2.填充下线的分桶到不可用列表中
            List<BucketCacheBO> offlineBucket = bucketLocalCache.getAvailableList().stream()
                    .filter(bucketCacheBO -> bucketCacheList.contains(bucketCacheBO.getBucketNo())).collect(Collectors.toList());

            if (CollectionUtils.isEmpty(offlineBucket)) {
                // 如果下线的分桶过滤后，没有要下线的了，直接返回
                return;
            }
            boolean isOfflineBucket = true;
            for (BucketCacheBO bucketCacheBO : offlineBucket) {
                if (!StringUtils.isEmpty(bucketCacheBO.getBucketNo())) {
                    bucketLocalCache.getUndistributedList().add(bucketCacheBO);
                    isOfflineBucket = false;
                }
            }
            // 有合法的下线分桶，才操作
            if (isOfflineBucket) {
                return;
            }
            // 过滤返回 还上线的分桶列表
            List<BucketCacheBO> availableList = bucketLocalCache.getAvailableList().stream()
                    .filter(bucketCacheBO -> !bucketCacheList.contains(bucketCacheBO.getBucketNo())).collect(Collectors.toList());
            bucketLocalCache.setAvailableList(availableList);

            //  可用分桶列表大于等于1的时候，才允许分桶下线
            if (availableList.size() >= 1) {
                // 元数据缓存更新，这里会切面处理，为避免出现延迟，先操作远程缓存的元数据覆盖

                bucketLocalCache.setOperationType(BucketStatusEnum.OFFLINE_STATUS.getCode());
                // 发一个消息进行本地缓存的更新，同时切面会发一个消息更新缓存，本地缓存的更新涉及版本号
                inventoryBucketCache.setBucketLocalCache(key, bucketLocalCache);

                // 分桶信息入库，待跑批任务执行
                inventoryRepository.saveBucketDetail(null, bucketLocalCache, BucketOperateEnum.OFFLINE.getCode(),
                        offlineBucket, null);
            }
        } catch (Exception e) {
            log.error("分桶下线出现失败", e);
            throw new BaseBizException(InventoryExceptionCode.INVENTORY_CACHE);
        } finally {
            inventoryBucketCache.threadLocalRemove();
        }
    }

    /**
     * 构建新的分桶元数据信息
     *
     * @param request    分桶上线对象
     * @param residueNum 中心桶剩余库存
     * @return
     */
    private void writeBucketCache(InventorOnlineRequest request, Integer residueNum) {
        String key = request.getSellerId() + request.getSkuId();
        //1. 获取到本地的缓存列表
        BucketLocalCache bucketLocalCache = inventoryBucketCache.getBucketLocalCache(key);
        try {
            if (!Objects.isNull(bucketLocalCache)) {
                // 获取当前可上线的分桶列表信息
                List<BucketCacheBO> bucketCacheBOList = buildBucketList(request.getBucketNoList(), bucketLocalCache.getAvailableList(), bucketLocalCache.getUndistributedList(), bucketLocalCache.getInventoryBucketConfig(), residueNum);
                // 当前可上线的分桶为空，直接返回
                if (CollectionUtils.isEmpty(bucketCacheBOList)) {
                    return;
                }

                // 中心桶被扣减掉的库存(上线的分桶库存总和)
                Integer descInventoryNum = bucketCacheBOList.stream().mapToInt(BucketCacheBO::getBucketNum).sum();

                // 构建返回新的元数据模型返回
                buildBucketLocalCache(bucketLocalCache, bucketCacheBOList, residueNum - descInventoryNum);

                // 分桶信息入库
                inventoryRepository.saveBucketDetail(null, bucketLocalCache, BucketOperateEnum.ONLINE.getCode(),
                        bucketCacheBOList, descInventoryNum);

                // 扣减中心桶，如果扣减失败了，直接抛异常
                // 中心桶的库存扣减信息
                Integer decr = tairCache.decr(buildSellerInventoryKey(request.getSellerId(), request.getSkuId()), descInventoryNum);
                if (decr < 0) {
                    throw new BaseBizException(InventoryExceptionCode.INVENTORY_CACHE);
                }
            }
        } catch (Exception e) {
            log.error("分桶构建初始化失败", e);
            throw new BaseBizException(InventoryExceptionCode.INVENTORY_CACHE);
        } finally {
            inventoryBucketCache.threadLocalRemove();
        }
    }

    /**
     * 分桶上线数据写入缓存
     *
     * @param inventoryBucketOperateBO
     */
    @Override
    public void processOnlineOperate(InventoryBucketOperateBO inventoryBucketOperateBO) {

//        log.error("执行分桶上线{}", JSONObject.toJSONString(inventoryBucketOperateBO));
        // 分桶元数据信息
        BucketLocalCache bucketLocalCache = JSON.parseObject(inventoryBucketOperateBO.getFeature(), BucketLocalCache.class);
        // 上线的分桶信息
        List<BucketCacheBO> bucketCacheBOList = JSON.parseArray(inventoryBucketOperateBO.getBucket(), BucketCacheBO.class);

        // 本次操作的id
        String operateId = inventoryBucketOperateBO.getOperateId();

        String key = buildBucketCacheKey(bucketLocalCache.getSellerId(), bucketLocalCache.getSkuId());
        // 上线失败的列表
        List<String> failBucketNos = Collections.synchronizedList(new ArrayList<>());
        CountDownLatch latch = new CountDownLatch(bucketCacheBOList.size());
        //1.先更新分桶的上线缓存处理操作
        for (BucketCacheBO bucketCacheBO : bucketCacheBOList) {
            executors.execute(() -> {
                boolean setFlag = tairCache.retrySet(bucketCacheBO.getBucketNo(), JSONObject.toJSONString(bucketCacheBO.getBucketNum()), 0, 3);
                if (!setFlag) {
                    failBucketNos.add(bucketCacheBO.getBucketNo());
                    // 记录失败，后续处理
                    inventoryRepository.saveFailOperate(operateId, TairInventoryConstant.OPERATE_INCR_FAIL,
                            bucketCacheBO.getBucketNo(), bucketCacheBO.getBucketNum());

                    tairCache.appendJsonExhset(TairInventoryConstant.OPERATE_INCR_FAIL,
                            bucketCacheBO.getBucketNo(), operateId, bucketCacheBO.getBucketNum());
                }
                latch.countDown();
            });
        }
        try {
            latch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        if (!CollectionUtils.isEmpty(failBucketNos)) {
            bucketLocalCache.getUndistributedList().addAll(failBucketNos.stream().map(BucketCacheBO::new).collect(Collectors.toList()));
            bucketLocalCache.setAvailableList(
                    bucketLocalCache.getAvailableList().stream()
                            .filter(bucketCacheBO -> failBucketNos.contains(bucketCacheBO.getBucketNo()))
                            .collect(Collectors.toList())
            );
        }
        //2.处理分桶列表的更新，待中心桶库存以及上线分桶库存更新完成，更新远程和本地的分桶列表
        bucketLocalCache.setOperationType(BucketStatusEnum.AVAILABLE_STATUS.getCode());
        inventoryBucketCache.setBucketLocalCache(key, bucketLocalCache);
    }

    /**
     * 构建新的元数据模型
     *
     * @param bucketLocalCache  本地分桶元数据信息
     * @param bucketCacheBOList 上线的分桶列表
     * @param residueNum        中心桶剩余库存
     */
    private void buildBucketLocalCache(BucketLocalCache bucketLocalCache, List<BucketCacheBO> bucketCacheBOList, Integer residueNum) {
        // 填充中心桶剩余库存
        bucketLocalCache.setResidueNum(residueNum);
        //添加新上线的分桶列表
        bucketLocalCache.getAvailableList().addAll(bucketCacheBOList);
        Map<String, BucketCacheBO> bucketCacheMap = bucketCacheBOList.stream()
                .collect(Collectors.toMap(BucketCacheBO::getBucketNo, Function.identity()));

        List<BucketCacheBO> undistributedList = bucketLocalCache.getUndistributedList().stream().filter(bucketCacheBO ->
                // 在上线的分桶列表，需要移除掉
                !bucketCacheMap.containsKey(bucketCacheBO.getBucketNo())).collect(Collectors.toList());
        // 从不可用的分桶列表重移除
        bucketLocalCache.setUndistributedList(undistributedList);
        bucketLocalCache.setOperationType(BucketStatusEnum.AVAILABLE_STATUS.getCode());
    }

    /**
     * 获取可上线的分桶列表信息以及具体上线库存
     *
     * @param bucketNoList            入参的上线分桶编号列表
     * @param availableList           上线正在使用的分桶编号列表
     * @param undistributedList       下线或者未使用的分桶编号列表
     * @param inventoryBucketConfigDO 当前分桶的配置模板信息
     * @param residueNum              中心桶的剩余可分配库存
     * @return 可上线的分桶列表以及具体分桶库存
     */
    private List<BucketCacheBO> buildBucketList(List<String> bucketNoList,
                                                List<BucketCacheBO> availableList,
                                                List<BucketCacheBO> undistributedList,
                                                InventoryBucketConfigDO inventoryBucketConfigDO,
                                                Integer residueNum) {
        //1. 如果入参选择了上线的分桶编号列表，则从缓存中配置的未使用分桶列表进行比对处理
        List<String> bucketCacheList = null;
        if (!CollectionUtils.isEmpty(bucketNoList)) {
            Map<String, BucketCacheBO> bucketCacheMap = undistributedList.stream()
                    .collect(Collectors.toMap(BucketCacheBO::getBucketNo, Function.identity()));

            // 过滤返回可用的分桶编号
            bucketCacheList = bucketNoList.stream()
                    .filter(bucketNo -> bucketCacheMap.containsKey(bucketNo))
                    .collect(Collectors.toList());
        } else {
            // 直接返回 下线的不可用分桶列表
            bucketCacheList = undistributedList.stream().map(BucketCacheBO::getBucketNo).collect(Collectors.toList());
        }
        // 可上线的分桶列表为空
        if (CollectionUtils.isEmpty(bucketCacheList)) {
            return Lists.newArrayList();
        }

        //2. 根据中心桶的可分配库存，处理返回具体上线的分桶配置信息
        return calcOnlineBucket(availableList, bucketCacheList, residueNum, inventoryBucketConfigDO);
    }


    /**
     * 构建上线的分桶库存模型
     *
     * @param availableList         上线正在使用的分桶编号列表
     * @param bucketCacheList       预上线的分桶列表
     * @param residueNum            中心桶剩余库存容量
     * @param inventoryBucketConfig 当前分桶配置信息
     * @return
     */
    private List<BucketCacheBO> calcOnlineBucket(List<BucketCacheBO> availableList,
                                                 List<String> bucketCacheList,
                                                 Integer residueNum,
                                                 InventoryBucketConfigDO inventoryBucketConfig) {
        List<BucketCacheBO> bucketCacheBOList = new ArrayList<>();
        // 获取已上线分桶 + 准备上线的分桶数量
        Integer sumBucketSize = availableList.size() + bucketCacheList.size();
        // 获取得到已上线的分桶分配库存深度总和
        int sumBucketNum = availableList.stream().mapToInt(BucketCacheBO::getBucketNum).sum();
        // 获取总的库存深度 + 中心桶的库存，得到平均的分桶实际可分配库存深度
        Integer averageNum = (sumBucketNum + residueNum) / sumBucketSize;
        // 当前准备分桶上线的数量(一般都是未使用的下线分桶)
        Integer bucketNum = bucketCacheList.size();
        // 计算一下平均分桶的库存 是否小于最小深度，如小于则以最小深度为准进行分桶
        Integer minBucketNum = bucketNum * averageNum;
        // 当库存数量小于最小分桶深度*分桶数量，减少可分配的分桶数量, 最后一个分桶分配剩余的全部库存(避免少量的库存分桶直接触发阈值下线)
        if (minBucketNum > residueNum) {
            bucketNum = residueNum / inventoryBucketConfig.getMinDepthNum();
            averageNum = inventoryBucketConfig.getMinDepthNum();
        }
        // 如果库存深度超过最大库存深度，则只存放最大深度
        if (averageNum > inventoryBucketConfig.getMaxDepthNum()) {
            averageNum = inventoryBucketConfig.getMinDepthNum();
        }
        // 当前没有准备分桶上线的数量
        if (bucketNum == 0) {
            return Lists.newArrayList();
        }

        // 开始填充每个分桶的具体上线库存
        for (int i = 0; i < bucketNum; i++) {
            BucketCacheBO bucketCache = new BucketCacheBO();
            // 这里上线的分桶数量不会超过实际能够上线的分桶数量，所以bucketCacheList.get(i)不会数组越界
            bucketCache.setBucketNo(bucketCacheList.get(i));
            bucketCache.setBucketNum(averageNum);
            bucketCache.setAllotNum(bucketCache.getBucketNum());
            bucketCacheBOList.add(bucketCache);
        }
        return bucketCacheBOList;
    }

    /**
     * 操作数据入缓存
     *
     * @param request
     */
    private void inventoryBucketCache(InventorBucketRequest request) {
        // 获取中心桶库存的key, 卖家+商品 sku 的库存, 中心桶缓存
        String key = buildSellerInventoryKey(request.getSellerId(), request.getSkuId());
        // 1. 先验证是否已有库存缓存存在
        BucketLocalCache bucketLocalCache = inventoryBucketCache.getBucketLocalCache(request.getSellerId() + request.getSkuId());
        // 缓存不存在，则走初始化操作
        try {
            if (Objects.isNull(bucketLocalCache)) {
                // 2.获取库存分桶的模板配置
                InventoryBucketConfigDO inventoryBucketConfig = inventoryRepository.getInventoryBucketConfig(request.getTemplateId());
                // 初始化分桶库存
                initInventoryBucket(request, inventoryBucketConfig);
            } else {
                // 3. 缓存已存在，直接把库存加到中心桶里面，并返回中心桶库存
                Integer residueNum = tairCache.incr(key, request.getInventoryNum());
                if (residueNum < 0) {
                    throw new BaseBizException(InventoryExceptionCode.INVENTORY_CACHE);
                }
                // 4 尝试将库存分配到新的分桶上（注意，先将中心桶的库存加上去）
                InventorOnlineRequest onlineRequest = inventoryConverter.converterRequest(request);
                // 5.构建新的分桶元数据信息，并写入
                writeBucketCache(onlineRequest, residueNum);
            }
        } catch (Exception e) {
            log.error("分桶库存初始化出现失败", e);
            throw new BaseBizException(InventoryExceptionCode.INVENTORY_CACHE);
        } finally {
            inventoryBucketCache.threadLocalRemove();
        }
    }

    /**
     * 初始化分桶库存
     *
     * @param request
     * @param inventoryBucketConfig
     */
    private void initInventoryBucket(InventorBucketRequest request, InventoryBucketConfigDO inventoryBucketConfig) {
        // 4.计算出分桶的数据信息
        // 根据分桶配置先计算出来, 要分多少个桶, 每个桶多少库存, 形成库存分桶元数据
        BucketLocalCache bucketLocalCache = calcInventoryBucket(request, inventoryBucketConfig);

        String incrementKey = TairInventoryConstant.SELLER_SKU_STOCK_COUNT_PREFIX + request.getSellerId() + request.getSkuId();

        // 如果没有，此时需要初始化一个
        if (!segmentNoCache.containsKey(incrementKey)) {
            synchronized (incrementKey.intern()) {
                if (!segmentNoCache.updateCacheFromDb(incrementKey)) {
                    // 生成一个库存的序列分段号
                    leafAllocNoRepository.insertLeadAlloc(incrementKey, 10000);
                    segmentNoCache.updateCacheFromDb(incrementKey);
                }
            }
        }

        // 分桶信息入库
        inventoryRepository.saveBucketDetail(null, bucketLocalCache, BucketOperateEnum.INIT.getCode(),
                bucketLocalCache.getAvailableList(), null);

        // 写入中心桶的库存信息
//        log.error("中心桶库存：{}",bucketLocalCache.getResidueNum());

        // 获取中心桶库存的key
        String key = buildSellerInventoryKey(bucketLocalCache.getSellerId(), bucketLocalCache.getSkuId());
        boolean setFlag = tairCache.set(key, bucketLocalCache.getResidueNum().toString(), 0);
        if (!setFlag) {
            // 中心桶库存写入失败，回滚事务
            throw new BaseBizException(InventoryExceptionCode.INVENTORY_CACHE);
        }
    }


    /**
     * 将初始化数据写入远程缓存以及本地缓存
     *
     * @param inventoryBucketOperateBO
     */
    @Override
    public void processInitBucket(InventoryBucketOperateBO inventoryBucketOperateBO) {
        BucketLocalCache bucketLocalCache = JSON.parseObject(inventoryBucketOperateBO.getFeature(), BucketLocalCache.class);
//        log.error("执行库存初始化{}", JSONObject.toJSONString(inventoryBucketOperateBO));
        // 卖家ID+商品skuId标识
        String key = buildBucketCacheKey(bucketLocalCache.getSellerId(), bucketLocalCache.getSkuId());
        //2. 写入数据到对应的缓存上
        List<BucketCacheBO> availableList = bucketLocalCache.getAvailableList();
        CountDownLatch latch = new CountDownLatch(availableList.size());
        List<String> failBucketNos = Collections.synchronizedList(new ArrayList<>());
        for (BucketCacheBO bucketCacheBO : availableList) {
            executors.execute(() -> {
//                log.error("bucketNo:{}, inventoryNum:{}",bucketCacheBO.getBucketNo(),bucketCacheBO.getBucketNum());
                boolean setFlag = tairCache.retrySet(bucketCacheBO.getBucketNo(),
                        JSONObject.toJSONString(bucketCacheBO.getBucketNum()), 0, 3);
//                log.error("set bucket inventory, bucketNo:{}, inventoryNum:{}, successful:{}",
//                        bucketCacheBO.getBucketNo(),bucketCacheBO.getBucketNum(),setFlag);
                if (!setFlag) {
                    failBucketNos.add(bucketCacheBO.getBucketNo());

                    // 记录失败，后续处理
                    tairCache.appendJsonExhset(TairInventoryConstant.OPERATE_SET_FAIL, bucketCacheBO.getBucketNo(),
                            inventoryBucketOperateBO.getOperateId(), bucketCacheBO.getBucketNum());

                    inventoryRepository.saveFailOperate(inventoryBucketOperateBO.getOperateId(),
                            TairInventoryConstant.OPERATE_SET_FAIL, bucketCacheBO.getBucketNo(), bucketCacheBO.getBucketNum());
                }
                latch.countDown();
            });
        }
        try {
            latch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        if (!CollectionUtils.isEmpty(failBucketNos)) {
            bucketLocalCache.getUndistributedList().addAll(failBucketNos.stream().map(BucketCacheBO::new).collect(Collectors.toList()));
            availableList = availableList.stream()
                    .filter(bucketCacheBO -> failBucketNos.contains(bucketCacheBO.getBucketNo()))
                    .collect(Collectors.toList());
            bucketLocalCache.setAvailableList(availableList);
        }
        //3. 记录存储到本地缓存列表
        bucketLocalCache.setOperationType(BucketStatusEnum.AVAILABLE_STATUS.getCode());
        inventoryBucketCache.setBucketLocalCache(key, bucketLocalCache);
//        log.error("元数据信息:{}",JSONObject.toJSONString(bucketLocalCache));
    }

    /**
     * 计算出当前库存入库的具体分桶信息
     *
     * @param request               分桶请求参数
     * @param inventoryBucketConfig 库存分桶配置信息
     */
    private BucketLocalCache calcInventoryBucket(InventorBucketRequest request, InventoryBucketConfigDO inventoryBucketConfig) {
        // 默认的分桶数量
        Integer bucketNum = inventoryBucketConfig.getBucketNum();
        // 总的库存数量
        Integer inventoryNum = request.getInventoryNum();
        // 分桶最大库存容量
        Integer maxBucketNum = bucketNum * inventoryBucketConfig.getMaxDepthNum();
        // 所有分桶最小的库存容量，如果需要放入分桶的库存数量低于这个值，那么只会分配给部分分桶，需要重新计算分桶
        Integer minBucketNum = bucketNum * inventoryBucketConfig.getMinDepthNum();
        // 当前需要放入分桶的库存数量
        int countBucketNum = Math.min(inventoryNum, maxBucketNum);

        // 当库存数量小于最小分桶深度*分桶数量，减少可分配的分桶数量, 最后一个分桶分配剩余的全部库存(避免少量的库存分桶直接触发阈值下线)
        if (minBucketNum > countBucketNum) {
            bucketNum = countBucketNum / inventoryBucketConfig.getMinDepthNum();
            // 如果库存数量不足一个分桶的最小深度，但是大于0，则上线一个分桶
            if (bucketNum == 0 && countBucketNum % inventoryBucketConfig.getMinDepthNum() > 0) {
                bucketNum++;
            }
        }

        // 获取每个分桶分配的库存数量
        Integer bucketInventoryNum = countBucketNum / bucketNum;
        // 剩余库存数量，可能为0或者大于0，补到最后一个分桶上。
        Integer residueNum = countBucketNum - bucketInventoryNum * bucketNum;
        // 2. 构建缓存数据模型, 以卖家ID+商品skuId为唯一标识
        String key = request.getSellerId() + request.getSkuId();

        BucketLocalCache bucketLocalCache = buildBucketCache(key, bucketNum, bucketInventoryNum, residueNum, inventoryBucketConfig);
        // 标记到具体的数据上
        bucketLocalCache.setSellerId(request.getSellerId());
        bucketLocalCache.setSkuId(request.getSkuId());
        bucketLocalCache.setInventoryNum(inventoryNum);
        // 中心桶剩余库存 = 总库存 - 当前需要放入分桶的库存数量
        bucketLocalCache.setResidueNum(inventoryNum - countBucketNum);
        bucketLocalCache.setInventoryBucketConfig(inventoryBucketConfig);
        bucketLocalCache.setOperationType(BucketStatusEnum.AVAILABLE_STATUS.getCode());
        return bucketLocalCache;
    }


    /**
     * 构建缓存模型
     *
     * @param key                   卖家ID+商品skuId为唯一标识
     * @param bucketNum             分桶数量
     * @param inventoryNum          分桶的库存数量
     * @param residueNum            剩余的未分配均匀的库存
     * @param inventoryBucketConfig 分桶配置信息
     */
    private BucketLocalCache buildBucketCache(String key, Integer bucketNum, Integer inventoryNum, Integer residueNum
            , InventoryBucketConfigDO inventoryBucketConfig) {
        // 先获取得到这个模板配置的对应可分槽位的均匀桶列表
        List<String> bucketNoList = InventorBucketUtil.createBucketNoList(key, inventoryBucketConfig.getBucketNum());

        // 可用的分桶, 情况: 分配的库存数量太少, 导致分桶数量是很少
        List<BucketCacheBO> bucketCacheBOList = new ArrayList<>(bucketNum);
        // 构建出多个分桶对象
        for (int i = 0; i < bucketNum; i++) {
            String bucketNo = bucketNoList.get(i);
            // 生成对应的分桶编号，方便定义到具体的分桶上
            BucketCacheBO bucketCache = new BucketCacheBO();
            bucketCache.setBucketNo(bucketNo);
            // 最后一个分桶，分配剩余未除尽的库存+平均库存
            if (i == bucketNum - 1) {
                bucketCache.setBucketNum(inventoryNum + residueNum);
            } else {
                bucketCache.setBucketNum(inventoryNum);
            }
            bucketCacheBOList.add(bucketCache);
        }

        // 不可用的分桶
        List<BucketCacheBO> undistributedList = new ArrayList<>(bucketNum);
        // 生成的分桶对象超过实际可分配的分桶对象，保留这批多余的分桶模型为不可用分桶，后续分桶上线可以选择使用
        if (bucketNoList.size() > bucketNum) {
            for (int i = bucketNum; i < bucketNoList.size(); i++) {
                BucketCacheBO bucketCache = new BucketCacheBO();
                String bucketNo = bucketNoList.get(i);
                bucketCache.setBucketNo(bucketNo);
                undistributedList.add(bucketCache);
            }
        }

        // 生成缓存的明细key
        List<String> bucketDetailKeyList = InventorBucketUtil.createBucketNoList(key, inventoryBucketConfig.getBucketNum(), "%07d");

        BucketLocalCache bucketLocalCache = new BucketLocalCache();
        // 设置分桶缓存明细的key
        bucketLocalCache.setBucketDetailKeyList(bucketDetailKeyList);
        // 设置可用的分桶缓存列表
        bucketLocalCache.setAvailableList(bucketCacheBOList);
        // 设置不可用或者已下线的分桶缓存列表
        bucketLocalCache.setUndistributedList(undistributedList);
        return bucketLocalCache;
    }

    /**
     * 获取卖家商品的分桶列表
     *
     * @param sellerId 商家 id
     * @param skuId    商品 sku id
     */
    private ProductInventoryDTO querySellerProductBucketList(String sellerId, String skuId) {

        String key = buildBucketCacheKey(sellerId, skuId);

        String bucketCache = inventoryBucketCache.getBucketCache(key);
//        log.error("远程缓存元数据：{}", bucketCache);
//        BucketLocalCache localCache = inventoryBucketCache.getBucketLocalCache(sellerId + skuId);
//        log.error("本地缓存元数据：{}", JSON.toJSONString(localCache));
        if (!StringUtils.isEmpty(bucketCache)) {
            // 引用缓存组件需要通过通用对象进行对应的缓存获取
            BucketLocalCache bucketLocalCache = JsonUtil.json2Object(bucketCache, BucketLocalCache.class);

            return buildProductInventoryDTO(bucketLocalCache);
        }

        return new ProductInventoryDTO();
    }

    /**
     * 构建查询对应的分桶信息
     *
     * @param bucketLocalCache
     * @return
     */
    private ProductInventoryDTO buildProductInventoryDTO(BucketLocalCache bucketLocalCache) {
        ProductInventoryDTO productInventory = new ProductInventoryDTO(bucketLocalCache.getSellerId(), bucketLocalCache.getSkuId());
        // 获取可用和不可用列表
        List<BucketCacheBO> availableList = bucketLocalCache.getAvailableList();
        List<BucketCacheBO> undistributedList = bucketLocalCache.getUndistributedList();
        if (!CollectionUtils.isEmpty(availableList)) {
            // 获取可用列表的分桶缓存集合key
            List<String> cacheKeyList = availableList.stream().map(BucketCacheBO::getBucketNo).collect(Collectors.toList());
            List<String> bucketNumList = tairCache.mget(cacheKeyList);

            List<BucketCacheDTO> bucketCacheDTOList = buildBucketCacheDTO(cacheKeyList, bucketNumList, availableList, BucketStatusEnum.AVAILABLE_STATUS.getCode());
            productInventory.setAvailableList(bucketCacheDTOList);
        }
        if (!CollectionUtils.isEmpty(undistributedList)) {
            // 获取下线列表的分桶缓存集合key
            List<String> cacheKeyList = undistributedList.stream().map(BucketCacheBO::getBucketNo).collect(Collectors.toList());
            List<String> bucketNumList = tairCache.mget(cacheKeyList);
            // 填充分桶详情信息
            List<BucketCacheDTO> bucketCacheDTOList = buildBucketCacheDTO(cacheKeyList, bucketNumList, availableList, BucketStatusEnum.OFFLINE_STATUS.getCode());
            productInventory.setUndistributedList(bucketCacheDTOList);
        }
        // 获取中心桶库存的key
        String sellerInventoryKey = buildSellerInventoryKey(bucketLocalCache.getSellerId(), bucketLocalCache.getSkuId());
        productInventory.setInventoryNum(tairCache.get(sellerInventoryKey));

        return productInventory;
    }

    /**
     * 填充分桶的详情信息
     *
     * @param cacheKeyList    分桶编号集合
     * @param bucketNumList   分桶库存集合
     * @param bucketCacheList 分桶详情列表
     * @param bucketStatus    分桶状态
     * @return
     */
    private List<BucketCacheDTO> buildBucketCacheDTO(List<String> cacheKeyList,
                                                     List<String> bucketNumList,
                                                     List<BucketCacheBO> bucketCacheList,
                                                     Integer bucketStatus) {
        Map<String, BucketCacheBO> bucketCacheBOMap = bucketCacheList.stream()
                .collect(Collectors.toMap(BucketCacheBO::getBucketNo, Function.identity()));
        List<BucketCacheDTO> bucketCacheDTOList = new ArrayList<>();
        for (int i = 0; i < cacheKeyList.size(); i++) {
            BucketCacheDTO bucketCacheDTO = new BucketCacheDTO();
            bucketCacheDTO.setBucketNo(cacheKeyList.get(i));
            if (StringUtils.isEmpty(bucketNumList.get(i))) {
                bucketCacheDTO.setInventoryNum(0);
            } else {
                bucketCacheDTO.setInventoryNum(Integer.valueOf(bucketNumList.get(i)));
            }
            bucketCacheDTO.setBucketStatus(bucketStatus);
            BucketCacheBO bucketCacheBO = bucketCacheBOMap.get(cacheKeyList.get(i));
            if (!Objects.isNull(bucketCacheBO)) {
                bucketCacheDTO.setAllotNum(bucketCacheBO.getAllotNum());
                bucketCacheDTO.setBucketNum(bucketCacheBO.getBucketNum());
            }
            bucketCacheDTOList.add(bucketCacheDTO);
        }
        return bucketCacheDTOList;
    }

    /**
     * 校验中心桶是否可上线分桶，并返回对应的可分配库存
     *
     * @param key
     * @return
     */
    private Integer checkBucketOnlineNum(String key) {
        // 验证缓存是否存在
        String residueNum = tairCache.get(key);
        if (StringUtils.isEmpty(residueNum)) {
            return 0;
        }
        // 返回具体的库存
        return Integer.valueOf(residueNum);
    }

    /**
     * 获取元数据的缓存key
     *
     * @param sellerId
     * @param skuId
     * @return
     */
    private String buildBucketCacheKey(String sellerId, String skuId) {
        return TairInventoryConstant.SELLER_BUCKET_PREFIX + sellerId + skuId;
    }

    /**
     * 获取锁的key
     *
     * @param sellerId
     * @param skuId
     * @return
     */
    private String buildBucketLockKey(String sellerId, String skuId) {
        return TairInventoryConstant.SELLER_INVENTORY_LOCK_PREFIX + sellerId + skuId;
    }

    /**
     * 获取下线锁的key
     *
     * @param sellerId
     * @param skuId
     * @param bucketNo
     * @return
     */
    private String buildBucketOfflineLockKey(String sellerId, String skuId, String bucketNo) {
        return TairInventoryConstant.SELLER_INVENTORY_LOCK_PREFIX + sellerId + skuId + bucketNo;
    }

    /**
     * 中心桶库存的key
     *
     * @param sellerId
     * @param skuId
     * @return
     */
    private String buildSellerInventoryKey(String sellerId, String skuId) {
        return TairInventoryConstant.SELLER_INVENTORY_PREFIX + sellerId + skuId;
    }

    /**
     * 卖家的商品 分桶下线创建时间的key
     *
     * @param sellerId
     * @param skuId
     * @return
     */
    private String buildSellerInventoryCreateTimeKey(String sellerId, String skuId, String bucketNo) {
        return TairInventoryConstant.SELLER_BUCKET_OFFLINE_CREATE_TIME_PREFIX + sellerId + skuId + bucketNo;
    }

    /**
     * 卖家的商品 分桶下线阈值
     *
     * @param sellerId
     * @param skuId
     * @return
     */
    private String buildSellerInventoryThresholdKey(String sellerId, String skuId) {
        return TairInventoryConstant.SELLER_INVENTORY_THRESHOLD_PREFIX + sellerId + skuId;
    }

    /**
     * 验证入参必填字段
     *
     * @param request
     */
    private void checkInventorParams(InventorBucketRequest request) {
        if (Objects.isNull(request)) {
            throw new BaseBizException(InventoryExceptionCode.PARAM_CHECK_ERROR, InventoryExceptionCode.PARAM_CHECK_ERROR.getErrorCode());
        }
        if (Objects.isNull(request.getInventorCode()) || Objects.isNull(request.getInventoryNum())
                || Objects.isNull(request.getSellerId()) || Objects.isNull(request.getSkuId())) {
            throw new BaseBizException(InventoryExceptionCode.PARAM_CHECK_ERROR, InventoryExceptionCode.PARAM_CHECK_ERROR.getErrorCode());
        }
        if (request.getInventoryNum() <= 0) {
            throw new BaseBizException(InventoryExceptionCode.PARAM_CHECK_ERROR, InventoryExceptionCode.PARAM_CHECK_ERROR.getErrorCode());
        }
    }

    /**
     * 验证入参必填字段
     *
     * @param request
     */
    private void checkInventorOnlineParams(InventorOnlineRequest request) {
        if (Objects.isNull(request)) {
            throw new BaseBizException(InventoryExceptionCode.PARAM_CHECK_ERROR, InventoryExceptionCode.PARAM_CHECK_ERROR.getErrorCode());
        }
        if (Objects.isNull(request.getSellerId()) || Objects.isNull(request.getSkuId())) {
            throw new BaseBizException(InventoryExceptionCode.PARAM_CHECK_ERROR, InventoryExceptionCode.PARAM_CHECK_ERROR.getErrorCode());
        }
    }

    /**
     * 验证入参必填字段
     *
     * @param request
     */
    private void checkInventorOfflineParams(InventorOfflineRequest request) {
        if (Objects.isNull(request)) {
            throw new BaseBizException(InventoryExceptionCode.PARAM_CHECK_ERROR, InventoryExceptionCode.PARAM_CHECK_ERROR.getErrorCode());
        }
        if (Objects.isNull(request.getSellerId()) || Objects.isNull(request.getSkuId())) {
            throw new BaseBizException(InventoryExceptionCode.PARAM_CHECK_ERROR, InventoryExceptionCode.PARAM_CHECK_ERROR.getErrorCode());
        }
        if (CollectionUtils.isEmpty(request.getBucketNoList())) {
            throw new BaseBizException(InventoryExceptionCode.PARAM_CHECK_ERROR, InventoryExceptionCode.PARAM_CHECK_ERROR.getErrorCode());
        }
    }
}
